package com.nanohadoop.mapreduce;

import com.nanohadoop.hdfs.HDFSClient;
import com.nanohadoop.yarn.ApplicationMaster;
import com.nanohadoop.yarn.YARNClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;

/**
 * MapReduce作业类，负责提交和管理MapReduce作业
 */
public class MapReduceJob {
    private static final Logger log = LoggerFactory.getLogger(MapReduceJob.class);
    private final JobConfiguration conf;
    private final YARNClient yarnClient;
    private String jobId;
    private ApplicationMaster appMaster;
    private Thread executionThread;

    public MapReduceJob(JobConfiguration conf, YARNClient yarnClient) {
        this.conf = conf;
        this.yarnClient = yarnClient;
    }

    /**
     * 提交作业到YARN集群
     */
    public void submit() throws IOException {
        // 验证配置
        validateConfiguration();

        // 提交作业到YARN
        this.jobId = yarnClient.submitApplication(conf);
        log.debug("Job submitted with ID: " + jobId);

        // 获取ApplicationMaster引用
        this.appMaster = yarnClient.getApplicationMaster(jobId);

        // 启动执行线程
        startExecution();
    }

    /**
     * 启动作业执行
     */
    private void startExecution() {
        executionThread = new Thread(() -> {
            try {
                // 创建HDFS客户端
                HDFSClient hdfsClient = new HDFSClient("localhost", 9000);

                // 创建执行器并执行作业
                MapReduceExecutor executor = new MapReduceExecutor(conf, hdfsClient, appMaster);
                executor.execute();
            } catch (Exception e) {
                e.printStackTrace();
                if (appMaster != null) {
                    appMaster.setStatus(ApplicationMaster.ApplicationStatus.FAILED);
                }
            }
        });
        executionThread.start();
    }

    /**
     * 验证作业配置
     */
    private void validateConfiguration() throws IOException {
        if (conf.getInputPath() == null || conf.getInputPath().isEmpty()) {
            throw new IOException("Input path is required");
        }
        if (conf.getOutputPath() == null || conf.getOutputPath().isEmpty()) {
            throw new IOException("Output path is required");
        }
        if (conf.getMapperClass() == null || conf.getMapperClass().isEmpty()) {
            throw new IOException("Mapper class is required");
        }
        if (conf.getReducerClass() == null || conf.getReducerClass().isEmpty()) {
            throw new IOException("Reducer class is required");
        }
    }

    /**
     * 检查作业是否已完成
     */
    public boolean isComplete() {
        if (appMaster == null) {
            return false;
        }
        return appMaster.getStatus() == ApplicationMaster.ApplicationStatus.COMPLETED ||
                appMaster.getStatus() == ApplicationMaster.ApplicationStatus.FAILED ||
                appMaster.getStatus() == ApplicationMaster.ApplicationStatus.KILLED;
    }

    /**
     * 获取作业状态
     */
    public ApplicationMaster.ApplicationStatus getStatus() {
        if (appMaster == null) {
            return null;
        }
        return appMaster.getStatus();
    }

    /**
     * 等待作业完成
     */
    public void waitForCompletion() throws InterruptedException {
        while (!isComplete()) {
            Thread.sleep(1000); // 每秒检查一次
        }
    }

    // Getters
    public String getJobId() { return jobId; }
    public JobConfiguration getConfiguration() { return conf; }
    public ApplicationMaster getApplicationMaster() { return appMaster; }
}